package com.atguigu.flink.chapter05.Transform;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 14:06
 */
public class Flink05_KeyBy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                    }
                });

        //TODO KeyBy
        // 如果使用 位置索引 或 字段名称 的方式来分组，那么返回的 key的类型为 通用类型 Tuple，使用的时候需要自己转换类型
        // 比如说，key为sensor的id
        //      Tuple => 具体的Tuple， Tuple1 => 取出，tuple1.f0

        // TODO 第一种方式： 指定 位置索引,只能用于 Tuple 的数据类型
//        KeyedStream<WaterSensor, Tuple> sensorKS = sensorDS.keyBy(0);

        // TODO 第二种方式：指定 字段名字,适用于 POJO
//        KeyedStream<WaterSensor, Tuple> sensorKS = sensorDS.keyBy("id");

        // TODO 第三种方式（推荐）：使用 KeySelector
        /*KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.getId();
            }
        });*/


        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
        sensorKS.print();


        env.execute();
    }

}
/*
    TODO 分组与分区的区别：
        分组： 是一个逻辑上的划分，按照key进行区分，  经过 keyby，同一个分组的数据肯定会进入同一个分区
        分区： 下游算子的一个并行实例（等价于一个slot），同一个分区内，可能有多个分组
 */

/*
    TODO keyby源码梳理：
        1. 对 key做了两次 hash ，并对 默认最大并行度 取, 得到一个 KeyGroupID：
                第一次, 调用自身的 hashCode方法；
                第二次，murmurHash;
                默认最大并行度 为 128;
        2. 计算 应该往 下游哪个分区（并行实例）发送：
                keyGroupId * parallelism / maxParallelism;
                两次哈希并取模的结果 * 下游算子的并行度 / 默认最大并行度128



    new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)
        DEFAULT_LOWER_BOUND_MAX_PARALLELISM = 2 << 7 = 128

    分区器里 => selectChannel()
        KeyGroupRangeAssignment.assignKeyToParallelOperator(key, maxParallelism, numberOfChannels);
            maxParallelism = 128
            numberOfChannels = 下游算子的 并行度（分区数）

     computeOperatorIndexForKeyGroup(maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));
        => keyGroupId * parallelism / maxParallelism;
            两次哈希并取模的结果 * 下游算子的并行度 / 默认最大并行度128

        => keyGroupId = assignToKeyGroup(key, maxParallelism)
            => computeKeyGroupForKeyHash(key.hashCode(), maxParallelism)        // 对 key 取了一个hash值
                => MathUtils.murmurHash(keyHash) % maxParallelism               // 对 key的hash值 再进行一次 murmur哈希
 */